package com.gate.ws.spring.rocketmq.consumer;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.gate.ws.spring.cache.LocalChannelCache;
import com.gate.ws.spring.utils.JsonUtils;
import com.nami.gate.api.request.message.DownMsgBatchReq;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

import static com.nami.basic.constants.DownTopicConst.FLUSH_MORE_TOPIC;


@Slf4j
@Component
@RocketMQMessageListener(topic = FLUSH_MORE_TOPIC, consumerGroup = "${rocketmq.consumer.flushMoreConsumerGroup}", selectorExpression = "${nami.mqTags}", consumeMode = ConsumeMode.CONCURRENTLY)
public class DownBatchMsgConsumer implements RocketMQListener<String> {


    @Override
    public void onMessage(String msg) {
        try {
            DownMsgBatchReq request = JsonUtils.parseObject(msg, DownMsgBatchReq.class);
            TimeInterval timer = DateUtil.timer();
            log.info("[flushMore] 批量发送 dto:{}", JsonUtils.toJsonString(request.getIdentityNos()));

            List<String> error = new ArrayList<>();

            List<String> identityNos = request.getIdentityNos();
            if (CollectionUtil.isEmpty(identityNos)) {
                return;
            }
            for (String identityNo : identityNos) {
                try {
                    Channel channel = LocalChannelCache.get(identityNo);
                    if (channel != null) {
                        log.error("[flushMore] 通道为null identityNo:{}", identityNo);
                        error.add(identityNo);
                        continue;
                    }
                    channel.writeAndFlush(request.getPacket());
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("[flushMore] 向[{}]发送消息失败:", identityNo, e.getMessage());
                    error.add(identityNo);
                } // end if catch exception
            } // end for loop

            log.warn("[flushMore] 群消息发送完毕 总数={}, 失败数：{}, use:{}ms", identityNos.size(), error.size(), timer.interval());
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
